package com.letv.dataprocess;

import java.sql.Connection;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;

import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.letv.bean.VideoBean;
import com.letv.common.ConnectionManager;
import com.letv.common.Constants;
import com.letv.common.ProcessData;
import com.letv.conf.ConfigFactory;
import com.letv.download.DownloadMapperService;
import com.letv.mms.cache.factory.CacheFactory;
import com.letv.util.DaquanAop;
import com.letv.util.DateUtil;
import com.letv.util.SQLUtil;

/**
 * @author GaoJiaWei
 *
 */
public class DataProcessWorker implements Callable<DataProcessResult> {

	private static final Logger log = LoggerFactory.getLogger(DataProcessWorker.class.getCanonicalName());

	private static String cbasekeyprefix = ConfigFactory.getString("cbaseprefix.name");
	
    // 外网取单视频流截图时的默认排序
    private static final String[] ARRAY_SORT = {"400*225",
                                                "320*180",
                                                "960*540",
                                                "1440*810",
                                                "400*300"};

    public DataProcessWorker(String tbName, String destTbName, String type, int begin, int end) {
    	this.tbName = tbName;
    	this.destTbName = destTbName;
    	this.type = type;
    	this.begin = begin;
    	this.end = end;
    }

	@Override
    public DataProcessResult call() throws Exception {
		return searchAndIntoDB(this.type);
	}

	private DataProcessResult searchAndIntoDB(String type) {
		DataProcessResult result = new DataProcessResult();
		log.info("DataProcessWorker call type:"+type);
		if (Constants.INDEX_LESO_EXTRAET.equals(type)) {
			log.info("DataProcessWorker call type:"+type);
			ConnectionManager cm = null;
			try {
				cm = ConnectionManager.getInstance();
			} catch (Exception e) {
				log.error("connection get instance error!", e);
				return result;
			}
	        Connection conn = null;
	        try {
	        	conn = cm.getConnectionFromLeso();
			} catch (Exception e) {
				log.error("get " + Constants.INDEX_LESO_EXTRAET + " id range error!", e);
				return result;
			}
	        ArrayList<Map<String, Object>> lists = new ArrayList<Map<String, Object>>();
	        int cbaseSuccessCnt=0;
	        int cbaseFailCnt=0;
	        try {
	            try {
	                lists = SQLUtil.getProcessNewLesoWebsiteData(conn, begin, end);
	            } catch(Exception e) {
	                log.error("temppoint error1" + e);
	            }
				for (Map<String, Object> record : lists) {
				    try {
	                    ArrayList<Map<String, Object>> videolists = null;
	                    String category = record.get("video_type").toString();
	                    HashMap<String, Object> record1 = new HashMap<String, Object>();
	                    String vid = record.get("id").toString();
	                    if (record.get("id")!=null && (!"".equals(record.get("id").toString()))) {
	                        log.info("website begin,video_id=" + vid);
	                        try {
	                            videolists = SQLUtil.getLesoAlbumVideoSourceData(conn, "leso.SQL3", vid,"180001");
	                        } catch(Exception e) {
	                            log.error("temppoint error2,vid=" + vid + e);
	                        }
	                        if( videolists ==null || videolists.size()  == 0) {
	                            log.info("single_video don't have data,video_id="+vid);
	                            continue;
	                        }
	                    }
	                    String tempKeyMark = cbasekeyprefix.indexOf("2") > 0 ? "2" : "";
	                    String oldkey = tempKeyMark + "leso_ea_" + record.get("leso_aid") + "_" + record.get("source");
	                    String key = cbasekeyprefix+"leso_ea_" + record.get("leso_aid") + "_" + record.get("source");
	                    String jsonStr = null;
	                    try {
	                        jsonStr = getWebSiteJSON(record, record1, videolists, category, key);
	                    } catch (Exception e) {
	                        log.error("jsonStr error,vid={}", vid, e);
	                    }
	                    List<String> daquanKeyList = new ArrayList<String>();
	                    if(Constants.IF_DAQUAN) {
	                        daquanKeyList.add(oldkey);
	                        daquanKeyList.add(key);
	                    } else {
	                        CacheFactory.getCache().set(oldkey, jsonStr, DateUtil.getIndexDayRefCurr(Calendar.HOUR, 48));
	                        CacheFactory.getCache().set(key, jsonStr, DateUtil.getIndexDayRefCurr(Calendar.HOUR, 48));
	                    }
	                    if ("letv".equals(record.get("source")) && record.get("vrs_aid") != null) {
	                        String key11 ="leso_la_" + record.get("vrs_aid");
	                        log.info(key+", website write key to cbase, leso_la_ key is " + tempKeyMark + key11);
	                        if(Constants.IF_DAQUAN) {
	                            daquanKeyList.add(tempKeyMark + key11);
	                            daquanKeyList.add(cbasekeyprefix + key11);
	                        } else {
	                            CacheFactory.getCache().set(tempKeyMark + key11, jsonStr, DateUtil.getIndexDayRefCurr(Calendar.HOUR, 48));
	                            CacheFactory.getCache().set(cbasekeyprefix + key11, jsonStr, DateUtil.getIndexDayRefCurr(Calendar.HOUR, 48));
	                        }
	                    }
	                    DaquanAop.addToDaquanLog(daquanKeyList, jsonStr, DaquanAop.DaquanOfflineMark.TOTAL_WEBSITE);
	                    cbaseSuccessCnt++;
	                    log.info("website write key to cbase, the key is " + key);
				    } catch(Exception e) {
				        log.error("temppoint error4,vid" + record.get("id").toString() + e);
				    }

				}
                log.info("website getProcessData, begin: " + begin + ",end: " + end + ",size:" + lists.size()
                        + ",put cabse size:" + cbaseSuccessCnt);
				result.setRecordsCnt(cbaseSuccessCnt);
				result.setCbaseSuccess(cbaseSuccessCnt);
				result.setCbasefail(cbaseFailCnt);
            } catch (Exception e) {
                log.error("temppoint error3,begin=" + begin + ",end=" + end + e);
	            e.printStackTrace();
	            result.setRecordsCnt(0);
				result.setFlag(false);
				result.setBegin(begin);
				result.setEnd(end);
            }finally {
                SQLUtil.close(conn);
			}
        }else{
			log.debug("current thread id=" + Thread.currentThread().getId() + " name:" + Thread.currentThread().getName() + " begin=" + begin + " end=" + end);
			ConnectionManager cm = null;
			try {
				cm = ConnectionManager.getInstance();
			} catch (Exception e) {
				log.error("connection get instance error!", e);
				result.setRecordsCnt(0);
				result.setCbaseSuccess(0);
				result.setCbasefail(0);
				result.setFlag(false);
				return result;
			}
	        Connection conn = cm.getConnection();
	        Connection connToMMS = cm.getConnectionFromMMS();
            try {
                processLeso(type, result, conn, connToMMS);
                log.info("current thread id=" + Thread.currentThread().getId() + " name:" + Thread.currentThread().getName() + " done!");
            } finally {
                SQLUtil.close(conn, connToMMS);
                log.info("processLeso local db/mms connections closed!");
            }
        }
		return result;
	}

    public static String getWebSiteJSON(Map<String, Object> record,
            HashMap<String, Object> record1,
            ArrayList<Map<String, Object>> videolists, String category,String key) {
        StringBuffer tempUrls = new StringBuffer();
        StringBuffer tempIntro = new StringBuffer("");
        StringBuffer deadlinks = new StringBuffer("");
        StringBuffer globalVid = new StringBuffer("");
        StringBuffer midsb = new StringBuffer("");
        StringBuffer isPaysb = new StringBuffer("");
        StringBuffer downloadPlatformb = new StringBuffer("");
        StringBuffer payPlatformb = new StringBuffer("");
        StringBuffer tempImages = new StringBuffer("");
        StringBuffer external_ids = new StringBuffer("");
        StringBuffer external_play_ids = new StringBuffer("");
        StringBuffer videoTypes = new StringBuffer("");
        boolean desc = category.equals("4") || category.equals("16")||category.equals("101");  //数据库为作品库
        int k = desc?videolists.size()-1:0;
        int end = desc?0:videolists.size()-1;
        boolean isLetv = Constants.WEBSITE_LETV.equals(record.get("source").toString());
        do {
            Map<String, Object> transientMap = videolists.get(k);
            String singleImage =null;
            try {
                if (transientMap.get("vImages") != null) {
                    JSONObject json = JSON.parseObject(transientMap.get("vImages").toString());
                    for (int as = 0; as < ARRAY_SORT.length; as++) {
                        if (json.containsKey(ARRAY_SORT[as])) {
                            singleImage = json.getString(ARRAY_SORT[as]);
                            break;
                        }
                    }
                }
            } catch (Exception e) {
                log.error("images get failed" + e);
            }
            String url = transientMap.get("play_link").toString();
            String video_external_id = (null == transientMap.get("external_id")) ? "" : transientMap.get("external_id").toString();
            String video_external_play_id = (null == transientMap.get("external_play_id")) ? "" : transientMap.get("external_play_id").toString();
            String singlevideo_videoType = (null == transientMap.get("type")) ? "" : transientMap.get("type").toString();
            Object episodetmp = transientMap.get("episode");
            String tmpDeadLink = transientMap.get("deadlinks").toString();
            Object tempGlobalVid = transientMap.get("global_vid");
            Object mid = transientMap.get("mid");
            strAppend(tempUrls, episodetmp, url,";");
            if (desc) {
                Object subName = null;
                if (isLetv) {
                    subName = transientMap.get("vsubName");
                }
                if (subName == null || "".equals(subName.toString().trim())) {
                    subName = transientMap.get("name");
                }
                DownloadMapperService.strAppend(tempIntro, episodetmp, subName,";");
            }
            strAppend(deadlinks, episodetmp, tmpDeadLink,";");
            strAppend(external_ids, episodetmp, video_external_id,";");
            strAppend(external_play_ids, episodetmp, video_external_play_id,";");
            strAppend(videoTypes, episodetmp, singlevideo_videoType,";");
            strAppend(tempImages, episodetmp, singleImage,";");
            strAppend(globalVid, episodetmp, tempGlobalVid,";");
            strAppend(isPaysb, episodetmp, transientMap.get("is_pay_cms"),";");
            strAppend(downloadPlatformb, episodetmp, transientMap.get("download_platform"),";");
            strAppend(payPlatformb, episodetmp, transientMap.get("pay_platform"),";");
            midsb.append(mid==null?"":(mid.toString().trim().length()>0?mid.toString().trim()+";":""));
            if (desc) {
                k--;
            } else {
                k++;
            }
        } while ( desc ? k >= end : k<=end );
        record1.put("videoPlayUrls", tempUrls.toString());
        record1.put("external_ids", external_ids.toString());
        record1.put("external_play_ids", external_play_ids.toString());
        record1.put("videoTypes", videoTypes.toString());
        record1.put("videoDeadlinkUrls", deadlinks.toString());
        record1.put("videoSingleVideoImages", tempImages.toString());
        record1.put("unIds", globalVid.toString());
        record1.put("intro", category.equals("4") || category.equals("16")?tempIntro.toString():"");
        record.put("mids", midsb.toString());
        record1.put("isPays", isPaysb.toString());
        record1.put("downloadPlatform", downloadPlatformb.toString());
        record1.put("payPlatform", payPlatformb.toString());

        record1.put("site", record.get("source"));
        record1.put("aid", record.get("leso_aid"));
        record1.put("vrsAid", record.get("vrs_aid"));
        record1.put("orderNum", record.get("top_hot"));
        record1.put("isPay", record.get("is_pay"));
        String is_cibn_checked = (null == record.get("is_cibn_checked")) ? "0" : record.get("is_cibn_checked").toString();
        record1.put("is_cibn_checked", is_cibn_checked);
        String album_external_id = (null == record.get("external_id")) ? "" : record.get("external_id").toString();
        record1.put("external_id", album_external_id);
        String album_external_play_id = (null == record.get("external_play_id")) ? "" : record.get("external_play_id").toString();
        if(StringUtils.isBlank(album_external_play_id)) {
            record1.put("external_play_id", new JSONObject());
        } else {
            try {
                record1.put("external_play_id", JSON.parseObject(album_external_play_id));
            } catch(Exception album_external_play_id_error) {
                record1.put("external_play_id", new JSONObject());
            }
        }
        String pay_platform = (null == record.get("pay_platform")) ? "" : record.get("pay_platform").toString();
        record1.put("pay_platform", pay_platform);
        int topHot =Integer.valueOf(record.get("top_hot").toString());
        int top_hot_tv = Integer.valueOf(record.get("top_hot_tv").toString());
        if("1".equals(record.get("is_pay").toString()) && category.equals("2")) {
            record1.put("isPc", "1");
            if(Constants.WEBSITE_LETV.equals(record.get("source").toString())) {
                record1.put("isTv", "1");
                record1.put("isMobile", "1");
            } else {
                record1.put("isTv", "0");
                record1.put("isMobile", "0");
                //topHot = record.get("top_hot")!=null && "1".equals(record.get("top_hot").toString())? 99 : 199;
            }
        } else {
            record1.put("isPc", "1");
            record1.put("isTv", "1");
            record1.put("isMobile", "1");
            //topHot = record.get("top_hot") == null ? 1 : Integer.valueOf(record.get("top_hot").toString());
        }
        //从作品库映射到媒资类型（Category）
         if (category.equals(String.valueOf(Constants.LESO_TV)) || category.equals(String.valueOf(Constants.LESO_MOVIE)) 
        		 || category.equals(String.valueOf(Constants.LESO_DONGMAN)) || category.equals(String.valueOf(Constants.LESO_ZONGYI))) 
         {
        	 category = Mappings.LESOTOLETV_MAPIING.get(category);
         }
        record1.put("category", category);
        record1.put("playStreams", record.get("quality"));
        record1.put("top_hot_def", "" + topHot);
        record1.put("top_hot_tv", "" + top_hot_tv);
        record1.put("orderNum", "" + topHot);
        ProcessData.setEpisodes(record1,record,videolists);
        record1.put("episodeStatus", record.get("episode_status")==null?"":record.get("episode_status").toString());
        record1.put("pushflag", record.get("pushflag")==null ? "" : record.get("pushflag").toString());
        record1.put("nowEpisodes", record.get("currentepisode"));
        record1.put("dataType", String.valueOf(Constants.ALBUM));

        Object playurl = record1.get("videoPlayUrls");
        Object deadLink = record1.get("videoDeadlinkUrls");
        Object singleVideoImage = record1.get("videoSingleVideoImages");
        String mids = (String) record.get("mids");
        String site = record.get("source").toString();
        if (null != playurl) {
            String videoPlayUrls = playurl.toString();
            List<VideoBean> videoList = new ArrayList<VideoBean>();
            String[] videoArray = videoPlayUrls.split(";");
            String[] info = record1.get("intro")!=null ? record1.get("intro").toString().split(";"): new String[0];
            String[] isPays = record1.get("isPays").toString().split(";");
            String[] payPlatforms = record1.get("payPlatform").toString().split(";");
            String[] downloadPlatforms = record1.get("downloadPlatform").toString().split(";");
            String playStreams = record.get("play_stream")!=null ? record.get("play_stream").toString() : "";
            String[] deadLinkArray = null;
            String[] singleVideoImageArray = null;
            if (null!=deadLink) {
                deadLinkArray = deadLink.toString().split(";");
            }
            if (null != singleVideoImage) {
                singleVideoImageArray = singleVideoImage.toString().split(";");
            }
            Map<Integer, String> map = new HashMap<Integer, String>();
            if (StringUtils.isNotEmpty(mids)) {
                String[] midsArray = mids.split(";");
                for (int i=0;i<midsArray.length; i++) {
                    map.put(i, midsArray[i]);
                }
            }
            for (int i = 0; i < videoArray.length; i++) {
                if (StringUtils.isNotBlank(videoArray[i])) {
                    String[] videoInfo = info.length > i ? info[i].split("\\|") : new String[0];
                    String[] ispay = isPays.length > i ? isPays[i].split("\\|") : new String[0];
                    String[] payPlatform = payPlatforms.length > i ? payPlatforms[i].split("\\|") : new String[0];
                    String[] downloadPlatform = downloadPlatforms.length > i ? downloadPlatforms[i].split("\\|") : new String[0];
                    String[] videoFileds = videoArray[i].split("\\|");
                    VideoBean video = new VideoBean();
                    video.setPlayStreams(playStreams); //添加playStream
                    if (videoFileds.length == 2) {
                        video.setDataType(Constants.VIDEO + "");
                        video.setAorder(StringUtils.isNotBlank(videoFileds[0])?videoFileds[0].replace("-", ""):"");
                        video.setUrl(videoFileds[1]);
                        video.setName(videoInfo.length > 1 ? videoInfo[1] : "精彩内容点击观看。");
                        video.setIsPay(ispay.length > 1 ? ispay[1] :null);
                        video.setPayPlatform(payPlatform.length > 1 ? payPlatform[1] :null);
                        video.setDownloadPlatform(downloadPlatform.length > 1 ? downloadPlatform[1] :null);
                        if (map.size()>0 && map.containsKey(i)) { //添加资源id
                            String mid = map.get(i).toString();
                            if (mid.contains(",")) {
                                mid = StringUtils.replace(mid, ",", "");
                            }
                            video.setMid(mid);
                        }else {
                            video.setMid("");
                        }
                        try {
                        if (StringUtils.isNotBlank(site) && "letv".equals(site)) {//添加vid
                            String url = video.getUrl();
                            int beginIndex = url.lastIndexOf("/");
                            int endIndex = url.lastIndexOf(".");
                            if(beginIndex < 0 || endIndex < 0) {
                                video.setVid("");
                            } else {
                                video.setVid(url.substring(beginIndex+1, endIndex));
                            }
                        } else {
                            video.setVid("");
                        }
                        } catch (Exception e) {
                            log.error("parse URL EROOR! url={}", video.getUrl(), e);
                        }
                        if (StringUtils.isNotBlank(videoFileds[1]) && videoFileds[1].contains("cloud.letv.com")) {//添加云盘id
                            String urlS = video.getUrl();
                            int len = urlS.lastIndexOf("/")+1;
                            video.setVideoPanid(urlS.substring(len));
                        } else {
                            video.setVideoPanid("");
                        }
                    }
                    if ((null!=deadLinkArray) && (videoArray.length == deadLinkArray.length)) {
                        if (StringUtils.isNotBlank(deadLinkArray[i])) {
                            String[] deadlinkFileds = deadLinkArray[i].split("\\|");
                            if (null!=deadlinkFileds && (deadlinkFileds.length==2)) {
                                video.setDeadlink(deadlinkFileds[1]); //每一集的deadlink
                            }
                        }
                    }else if(!record.get("source").equals("letv")){
                        log.info("website videoArray.length != deadLinkArray.length, the key is " + key);
                    }
                    if ((null != singleVideoImageArray) && (videoArray.length == singleVideoImageArray.length)) {
                        if (StringUtils.isNotBlank(singleVideoImageArray[i])) {
                            String[] singleVideoImageFileds = singleVideoImageArray[i].split("\\|");
                            if (null != singleVideoImageFileds && (singleVideoImageFileds.length == 2)) {
                                video.setSinglevideoImage(singleVideoImageFileds[1]);
                            }
                        }
                    }
                    videoList.add(video);
                }
            }
            if (videoList.size() > 0) {
                record1.put("videoList", videoList);
            }
        }
        removeTempProperty(record1);
        return JSONObject.toJSONString(record1);
    }

    private static void removeTempProperty(HashMap<String, Object> record1) {
        record1.remove("isPays");
        record1.remove("downloadPlatform");
        record1.remove("payPlatform");
    }

    private static void strAppend(StringBuffer tempUrls, Object episodetmp,
            Object string, String string2) {
        DownloadMapperService.strAppend(tempUrls, episodetmp, string,string2);
        
    }

    private void processLeso(String type, DataProcessResult result, Connection conn, Connection connToMMS) {
        ArrayList<Map<String, Object>> lists = new ArrayList<Map<String, Object>>();
        HashMap<String, Object> fieldNameMap = new HashMap<String, Object>();

        try {
        	lists = SQLUtil.getProcessData(conn, this.tbName, begin, end);
        	log.info("get datasize=" + lists.size());
        } catch (Exception e) {
        	log.error("get data type=" + type + " begin=" + begin + " end=" + end + " error!", e);
        }

        //取数据库字段和索引字段的映射，灌CBASE用
        try {
        	fieldNameMap = SQLUtil.getIndexFieldName(conn, "index_" + this.type);
        } catch (Exception e) {
        	log.error("get " + type + " index fieldname error!", e);
        }

        Map<String, Object> processResult = new HashMap<String, Object>();
        ArrayList<String> records = new ArrayList<String>();
        log.info("begin mapping " + type + " data!");

        processResult = processData(lists, this.type, fieldNameMap, connToMMS);
        records = (ArrayList<String>) processResult.get("records");
        try {
            if (records.size() > 0) {
                SQLUtil.insertIndexData(conn, type, records);
                log.info("insert " + type + " data, size=" + records.size());
                result.setRecordsCnt(records.size());
                result.setCbaseSuccess((Integer) processResult.get("cbaseSuccessCnt"));
                result.setCbasefail((Integer) processResult.get("cbaseFailCnt"));
                result.setFlag(true);
                result.setBegin(begin);
                result.setEnd(end);
        	} else {
        		log.info("datasize=0, no " + type + " data insert! flag=" + result.isFlag());
        	}
        } catch (Exception e) {
        	log.error("insert data type=" + type + " begin=" + begin + " end=" + end + " error:" + e.toString());
        	result.setRecordsCnt(0);
        	result.setFlag(false);
        	result.setBegin(begin);
        	result.setEnd(end);
        }
    }

	private Map<String, Object> processData(ArrayList<Map<String, Object>> lists, String type, HashMap<String, Object> f, Connection connToMMS) {
		Map<String, Object> processResult = new HashMap<String, Object>();
		ArrayList<String> records = new ArrayList<String>();
		DataprocessServiceImpl dsl = new DataprocessServiceImpl();

		boolean cbaseResult = false;
		int cbaseSuccessCnt = 0;
		int cbaseFailCnt = 0;

		if (Constants.INDEX_VIDEO.equals(type)) {
			cbaseSuccessCnt = 0;
			cbaseFailCnt = 0;
			for (int i = 0; i < lists.size(); i++) {
				Map<String, Object> map = lists.get(i);
				Map<String, Object> m = ProcessData.processVideo(map);
				String record = ProcessData.getIndexVideoStr(m);
				records.add(record);
				if (f.size() != 0) {
					Map<String, Object> indexMap = ProcessData.getCBaseMap(f, m, "video");

					cbaseResult = dsl.putCbase(Constants.SRC_VRS, indexMap, "all","video");
					if (cbaseResult) {
						cbaseSuccessCnt++;
					} else {
						cbaseFailCnt++;
					}
				} else {
					log.error("get " + type + " index fieldname error! put id=" + m.get("ID") + " into cbase failed!");
					cbaseFailCnt++;
				}
			}
		} else if (Constants.INDEX_LESO.equals(type)) {
			cbaseSuccessCnt = 0;
			cbaseFailCnt = 0;
			for (int i = 0; i < lists.size(); i++) {
				Map<String, Object> map = lists.get(i);
				Map<String, Object> m = ProcessData.processLeso(map);
	            Object vid= m.get("vid");
	            if ( vid != null && StringUtils.isNotBlank(vid.toString())) {
	                String record = ProcessData.getIndexLesoStr(m);
	                records.add(record);
	            }
				if (f.size() != 0) {
					Map<String, Object> indexMap = ProcessData.getCBaseMap(f, m, "leso");
					cbaseResult = dsl.putCbase(Constants.SRC_LESO, indexMap, "all","album");
					dsl.putLetvMappingToCbase(m);
					if (cbaseResult) {
						cbaseSuccessCnt++;
					} else {
						cbaseFailCnt++;
					}
				} else {
					log.error("get " + type + " index fieldname error! put id=" + m.get("ID") + " into cbase failed!");
					cbaseFailCnt++;
				}
			}
		} else if (Constants.INDEX_STAR.equals(type)) {
			cbaseSuccessCnt = 0;
			cbaseFailCnt = 0;
			for (int i = 0; i < lists.size(); i++) {
				Map<String, Object> map = lists.get(i);
				Map<String, Object> m =ProcessData.processStar(map);

				String record = ProcessData.getIndexStarStr(m);
				records.add(record);

				if (f.size() != 0) {
					Map<String, Object> indexMap = ProcessData.getCBaseMap(f, m, "star");
					cbaseResult = dsl.putCbase(Constants.SRC_VRS, indexMap, "all","star");
					if (cbaseResult) {
						cbaseSuccessCnt++;
					} else {
						cbaseFailCnt++;
					}
				} else {
					log.error("get " + type + " index fieldname error! put id=" + m.get("ID") + " into cbase failed!");
					cbaseFailCnt++;
				}
			}
		} else if (Constants.INDEX_ALBUM.equals(type)) {
			cbaseSuccessCnt = 0;
			cbaseFailCnt = 0;
			for (int i = 0; i < lists.size(); i++) {
				Map<String, Object> map = lists.get(i);
				Map<String, Object> m = ProcessData.processAlbum(map, connToMMS);
				//Map<String, Object> m = ProcessData.processAlbum(map);

				String record = ProcessData.getIndexAlbumStr(m);
				records.add(record);

				if (f.size() != 0) {
					Map<String, Object> indexMap = ProcessData.getCBaseMap(f, m, "album");

					cbaseResult = dsl.putCbase(Constants.SRC_VRS, indexMap, "all","album");
					if (cbaseResult) {
						cbaseSuccessCnt++;
					} else {
						cbaseFailCnt++;
					}
				} else {
					log.error("get " + type + " index fieldname error! put id=" + m.get("ID") + " into cbase failed!");
					cbaseFailCnt++;
				}
			}
		}

		processResult.put("records", records);
		processResult.put("cbaseFailCnt", cbaseFailCnt);
		processResult.put("cbaseSuccessCnt", cbaseSuccessCnt);

		return processResult;
	}

	public int getBegin() {
		return begin;
	}

	public void setBegin(int begin) {
		this.begin = begin;
	}

	public int getEnd() {
		return end;
	}

	public void setEnd(int end) {
		this.end = end;
	}

	public String getType() {
		return type;
	}

	public void setType(String type) {
		this.type = type;
	}

	public String getTbName() {
		return tbName;
	}

	public void setTbName(String tbName) {
		this.tbName = tbName;
	}

	public String getDestTbName() {
		return destTbName;
	}

	public void setDestTbName(String destTbName) {
		this.destTbName = destTbName;
	}

	public String getPushTarget() {
		return pushTarget;
	}

	public void setPushTarget(String pushTarget) {
		this.pushTarget = pushTarget;
	}

	private int begin;
	private int end;
	private String tbName;
	private String destTbName;
	private String type;
	private String pushTarget;
}